package cn.spark.study.streaming;

import com.google.common.base.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * 基于transform的实时广告计费日志黑名单过滤
 *
 * @author zhangj
 * @date 2020/11/19
 */
public class TransformBlackList {
	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("TransformBlackList");

		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

		//先模拟一份黑名单RDD
		List<Tuple2<String, Boolean>> blackList = new ArrayList<Tuple2<String, Boolean>>();
		blackList.add(new Tuple2<String, Boolean>("tom", true));
		final JavaPairRDD<String, Boolean> blackListRDD = jssc.sc().parallelizePairs(blackList);

		//日志格式(date,name)
		JavaReceiverInputDStream<String> adsClickLogDStream = jssc.socketTextStream("ymm1", 9999);

		//对输入的数据,进行下一步转换,变成(username,date username)
		JavaPairDStream<String, String> userAdsClickLogDStream = adsClickLogDStream.mapToPair(new PairFunction<String, String, String>() {
			@Override
			public Tuple2<String, String> call(String adsClickLog) throws Exception {
				return new Tuple2<String, String>(adsClickLog.split(" ")[1], adsClickLog);
			}
		});

		//可以执行transform操作了,将每个batch的RDD与黑名单RDD进行join,filter,map等操作
		JavaDStream<String> validAdsClickLogDStream = userAdsClickLogDStream.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
			@Override
			public JavaRDD<String> call(JavaPairRDD<String, String> userAdsClickLogRDD) throws Exception {
				//这里为什么用左外连接,并不是每个用户都在黑名单中
				//所以使用join那么没有存在于黑名单中的的数据,会无法Join到
				//所以这里用leftOuterJoin,就是说,哪怕一个user不在黑名单RDD中,没有Join到也还是会保存下来
				JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinedRDD = userAdsClickLogRDD.leftOuterJoin(blackListRDD);

				//连接之后执行filter算子
				JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filteredRDD =
						joinedRDD.filter(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() {
							@Override
							public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
								if (tuple._2._2.isPresent() && tuple._2._2().get()) {
									return false;
								}
								return true;
							}
						});

				//此时filteredRDD后,就剩下没有被黑名单过滤点击的用户了
				JavaRDD<String> validAdsClickLogRDD = filteredRDD.map(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, String>() {
					@Override
					public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
						return tuple._2()._1();
					}
				});

				return validAdsClickLogRDD;
			}
		});

		validAdsClickLogDStream.print();

		jssc.start();
		jssc.awaitTermination();
		jssc.close();
	}
}
